package com.erainm.custompartitioner;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by erainm on 2021/9/9 5:49 下午.
 *
 * @Description TODO:
 **/
public class MyProducer {
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<>();
        // 初始连接用到的broker地址
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "lg04:9092");
        // 指定key的序列化类
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 指定value的序列化类
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 指定自定义分区器类
        configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitioner.class);
        KafkaProducer<String, String> producer = new KafkaProducer<>(configs);

        // 这里不要设置partition的值
        ProducerRecord<String,String> record = new ProducerRecord<String,String>("tp_part_01","mykey","myvalue");

        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (null != exception) {
                    System.out.println("小修发送失败！");
                } else {
                    System.out.println(
                            "主题：" + metadata.topic() + "\t" +
                            "分区：" + metadata.partition() + "\t" +
                            "偏移量：" + metadata.offset()
                    );
                }
            }
        });
        producer.close();
    }
}
